package avicit.bdp.dcs.task.service;

import avicit.bdp.common.base.BaseService;
import avicit.bdp.common.dto.BdpFileStatus;
import avicit.bdp.common.dto.StorageResourceConf;
import avicit.bdp.common.dto.TaskInstanceDto;
import avicit.bdp.common.service.dto.BdpColumnDTO;
import avicit.bdp.common.service.dto.BdpTableDTO;
import avicit.bdp.common.service.dto.DataSourceDTO;
import avicit.bdp.common.service.service.DataQueryService;
import avicit.bdp.common.service.service.DataSourceService;
import avicit.bdp.common.utils.BdpLogUtil;
import avicit.bdp.common.utils.CheckParaUtils;
import avicit.bdp.common.utils.database.ColumnInfo;
import avicit.bdp.common.utils.database.DBUtils;
import avicit.bdp.common.utils.enums.JobType;
import avicit.bdp.common.utils.uploads.FileAdapterUtils;
import avicit.bdp.common.utils.uploads.IFileAdapter;
import avicit.bdp.core.constant.Constants;
import avicit.bdp.core.constant.ProcessConstants;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dcs.alert.dto.AlertDTO;
import avicit.bdp.dcs.alert.service.AlertService;
import avicit.bdp.dcs.datasource.service.BaseDbService;
import avicit.bdp.dcs.datasource.service.DataSourceUtils;
import avicit.bdp.dcs.detail.dto.TaskDataDetailDTO;
import avicit.bdp.dcs.detail.dto.TaskTableDetailDTO;
import avicit.bdp.dcs.detail.service.TaskDataDetailService;
import avicit.bdp.dcs.detail.service.TaskTableDetailService;
import avicit.bdp.dcs.detail.service.TaskTableLogService;
import avicit.bdp.dcs.job.dao.DcsJobDao;
import avicit.bdp.dcs.job.dto.JobDTO;
import avicit.bdp.dcs.job.dto.TableDto;
import avicit.bdp.dcs.semistructure.dto.CatalogAndFileDto;
import avicit.bdp.dcs.semistructure.dto.CatalogTemplateDTO;
import avicit.bdp.dcs.semistructure.dto.JobFileDTO;
import avicit.bdp.dcs.semistructure.service.CatalogTemplateService;
import avicit.bdp.dcs.semistructure.service.DcsFileService;
import avicit.bdp.dcs.semistructure.service.FileManageService;
import avicit.bdp.dcs.semistructure.service.JobFileService;
import avicit.bdp.dcs.task.dao.TaskDao;
import avicit.bdp.dcs.task.dto.ColumnMappingDto;
import avicit.bdp.dcs.task.dto.StreamColumnMappingDto;
import avicit.bdp.dcs.task.dto.TableMappingDto;
import avicit.bdp.dcs.task.dto.TaskDTO;
import avicit.bdp.dcs.tools.enums.*;
import avicit.bdp.dcs.utils.CommonUtil;
import avicit.bdp.dcs.utils.DcsConstants;
import avicit.bdp.dcs.utils.DcsErrorCode;
import avicit.platform6.api.sysmsg.dto.SysMsgDTO;
import avicit.platform6.api.system.SysMsgClient;
import avicit.platform6.commons.utils.ComUtil;
import avicit.platform6.core.context.ThreadContextHelper;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.*;
import java.util.stream.Collectors;

/**
 * @金航数码科技有限责任公司
 * @作者：liyb
 * @邮箱：liyb@avic-digital.com
 * @创建时间： 2020-08-10 18:53
 * @类说明： 任务服务类
 * @修改记录：
 */
@Service
public class TaskService extends BaseService<TaskDao, TaskDTO> {

  private static final Logger logger = LoggerFactory.getLogger(TaskService.class);

  @Autowired
  private DcsJobDao jobService;
  @Autowired
  private TaskTableDetailService taskTableDetailService;
  @Autowired
  private TaskDataDetailService taskDataDetailService;
  @Autowired
  private JobFileService jobFileService;
  @Autowired
  private DcsFileService fileService;
  @Autowired
  private TaskFactory taskFactory;
  @Autowired
  private CatalogTemplateService catalogTemplateService;
  @Autowired
  private BaseDbService baseDbService;
  @Autowired
  private DataQueryService dataQueryService;
  @Autowired
  private DataSourceService dataSourceService;
  @Autowired
  private TaskTableLogService tableLogService;
  @Autowired
  private FileManageService fileManageService;
  @Autowired
  private DataSourceUtils dataSourceUtils;
  @Autowired
  private AlertService alertService;
  @Autowired
  private SysMsgClient sysMsgClient;

  /**
   * 查看任务
   */
  public TaskDTO getTaskById(String jobId, String srcTableName) throws Exception {
    JobDTO jobDTO = jobService.selectByPrimaryKey(jobId);
    CheckParaUtils.checkDbObjectValidity(jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobId));

    TaskInterface taskInterface = taskFactory.getTaskInterface(jobDTO.getJobType());
    CheckParaUtils.checkObjectValidity(taskInterface, logger, DcsErrorCode.UNSUPPORTED_JOB_TYPE.getDesc() + jobDTO.getJobType());

    return taskInterface.getTaskById(jobDTO, srcTableName);
  }

  /**
   * 根据jobId查询增量、全量同步的表
   */
  public List<TableDto> getTableListByJobId(String jobId, Integer syncMode) {
    List<TaskDTO> taskDTOList = new ArrayList<>();
    TaskDTO taskQuery = new TaskDTO();
    taskQuery.setJobId(jobId);
    taskQuery.setFlag(Enable.ZERO.getCode());
    if (syncMode != null && syncMode == SyncMode.TOTAL_SYNC.getCode()) {
      taskQuery.setSyncMode(SyncMode.TOTAL_SYNC.getCode());
      taskDTOList = this.selectList(taskQuery);
    }else{
      // datax增量和通用增量
      taskQuery.setSyncMode(SyncMode.DATAX_INCREMENT_SYNC.getCode());
      taskDTOList = this.selectList(taskQuery);

      taskQuery.setSyncMode(SyncMode.COMMON_INCREMENT_SYNC.getCode());
      List<TaskDTO> taskList = this.selectList(taskQuery);
      taskDTOList.addAll(taskList);
    }

    // 减少传输到前端的数据量
    List<TableDto> tableList = new ArrayList<>();
    if (CollectionUtils.isNotEmpty(taskDTOList)) {
      taskDTOList.forEach(task -> {
        TableDto tableDto = new TableDto();
        tableDto.setId(task.getId());
        tableDto.setSrcTableName(task.getSrcTableName());
        tableDto.setDstTableName(task.getDstTableName());
        tableList.add(tableDto);
      });
    }
    return tableList;
  }

  /**
   * 更新任务
   */
  @Transactional(rollbackFor = Exception.class)
  public void updateTask(TaskDTO taskDTO) throws Exception {
    TaskDTO oldTask = this.selectByPrimaryKey(taskDTO.getId());
    CheckParaUtils.checkDbObjectValidity(oldTask, logger, String.format(DcsErrorCode.TASK_NULL.getDesc(), taskDTO.getId()));

    TaskInterface taskInterface = taskFactory.getTaskInterface(oldTask.getJobType());
    CheckParaUtils.checkObjectValidity(taskInterface, logger, DcsErrorCode.UNSUPPORTED_JOB_TYPE.getDesc() + oldTask.getJobType());

    String dataxJson = taskInterface.updateTask(oldTask, taskDTO);

    taskDTO.setDataxJson(dataxJson);
    this.updateByPrimaryKeySelective(taskDTO);
    BdpLogUtil.log4Update(taskDTO, oldTask);
    logger.info("更新任务成功,dataxJson:{}", dataxJson);
  }

  /**
   * 批量保存任务
   */
  @Transactional(rollbackFor = Exception.class)
  public void saveTasksByJob(String jobId) throws Exception {
    logger.info("保存任务开始，jobId:{}", jobId);
    // 查找job记录
    JobDTO jobDTO = jobService.selectByPrimaryKey(jobId);
    CheckParaUtils.checkDbObjectValidity(jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobId));

    TaskInterface taskInterface = taskFactory.getTaskInterface(jobDTO.getJobType());
    CheckParaUtils.checkObjectValidity(taskInterface, logger, DcsErrorCode.UNSUPPORTED_JOB_TYPE.getDesc() + jobDTO.getJobType());

    taskInterface.saveTasksByJobId(jobDTO);
    logger.info("保存任务结束，jobId:{}", jobId);
  }

  /**
   * 保存明细数据
   */
  public void saveTaskDetailData(String srcTable, String dstTable, TaskDTO taskDTO, JobDTO jobDTO) throws Exception {
    // 1、保存bdp_dcs_task_table_detail表数据
    boolean createTable = isCreateTable(jobDTO, dstTable);
    if (createTable) {
      TaskTableDetailDTO taskTableDetailDTO = TaskTableDetailDTO.builder()
              .id(ComUtil.getId())
              .taskId(taskDTO.getId())
              .srcTableName(srcTable)
              .dstTableName(dstTable)
              .taskStatus(TaskDetailStatus.NOT_STARTED.getCode()).build();
      taskTableDetailService.insert(taskTableDetailDTO);
    }

    // 2、保存bdp_dcs_task_data_detail表数据
    if ((jobDTO.getJobType() == JobType.TRANSFER.getCode() && jobDTO.getTransferType() != null
            && jobDTO.getTransferType().contains("2")) ||
            jobDTO.getJobType() == JobType.SYNC.getCode() ||
            jobDTO.getJobType() == JobType.SEMISTRUCTURE.getCode() ||
            jobDTO.getJobType() == JobType.STREAM.getCode() ||
            jobDTO.getJobType() == JobType.API_HTTP.getCode()) {

      TaskDataDetailDTO taskDataDetailDTO = TaskDataDetailDTO.builder()
              .id(ComUtil.getId())
              .taskId(taskDTO.getId())
              .srcTableName(srcTable)
              .dstTableName(dstTable)
              .taskStatus(TaskDetailStatus.NOT_STARTED.getCode()).build();
      taskDataDetailService.insert(taskDataDetailDTO);
    }
  }

  public ArrayList<ColumnMappingDto> getColumnMappingDtos(List<ColumnInfo> mergeColumnList, String datasource) {
    ArrayList<ColumnMappingDto> columnMappingDtos = new ArrayList<>();
    if (CollectionUtils.isNotEmpty(mergeColumnList)) {
      mergeColumnList.forEach(columnInfo -> {
        if (columnInfo.getLength() == null) {
          columnInfo.setLength(-1);
        }

        ColumnMappingDto columnMappingDto = ColumnMappingDto.builder()
                .srcColumnName(columnInfo.getName())
                .srcColumnType(columnInfo.getType())
                .srcColumnLength(columnInfo.getLength())
                .dstColumnName(columnInfo.getName(), datasource)
                .dstColumnType(columnInfo.getType())
                .dstColumnLength(columnInfo.getLength())
                .desensitizationPolicyId("")
                .desensitizationRuleId("").build();
        columnMappingDtos.add(columnMappingDto);
      });
    }
    return columnMappingDtos;
  }

  public ArrayList<ColumnMappingDto> getMergeColumnList(List<ColumnInfo> srcColumnList, List<ColumnInfo> dstColumnList) {
    ArrayList<ColumnMappingDto> columnMappingDtoList = new ArrayList<>();
    if (CollectionUtils.isNotEmpty(srcColumnList) && CollectionUtils.isNotEmpty(dstColumnList)) {
      for (ColumnInfo srcColumnInfo : srcColumnList) {
        for (ColumnInfo dstColumnInfo : dstColumnList) {

          // 列映射关系忽略大小写
          if (srcColumnInfo.getName().equalsIgnoreCase(dstColumnInfo.getName())) {
            if (srcColumnInfo.getLength() == null) {
              srcColumnInfo.setLength(-1);
            }
            if (dstColumnInfo.getLength() == null) {
              dstColumnInfo.setLength(-1);
            }

            ColumnMappingDto columnMappingDto = ColumnMappingDto.builder()
                    .srcColumnName(srcColumnInfo.getName())
//                                .srcColumnName(srcColumnInfo.getName().toLowerCase())
                    .srcColumnType(srcColumnInfo.getType())
                    .srcColumnLength(srcColumnInfo.getLength())
                    .dstColumnName(dstColumnInfo.getName(), "")
//                                .dstColumnName(dstColumnInfo.getName().toLowerCase(), "")
                    .dstColumnType(dstColumnInfo.getType())
                    .dstColumnLength(dstColumnInfo.getLength())
                    .desensitizationPolicyId("")
                    .desensitizationRuleId("").build();
            columnMappingDtoList.add(columnMappingDto);
          }
        }
      }
    }
    return columnMappingDtoList;
  }

  /**
   * 判断表在目标库中是否存在
   */
  public Boolean isExistInDstDatabase(String dstDatasourceId, String dstTable) throws Exception {
    return baseDbService.isExistTable(dstDatasourceId, dstTable);
  }

  /**
   * 验证语法
   */
  @Transactional
  public Boolean checkGrammar(String taskId, String filterCondition) throws Exception {
    // 过滤条件为空直接返回true
    if (StringUtils.isEmpty(filterCondition)) {
      return true;
    }

    TaskDTO taskDTO = this.selectByPrimaryKey(taskId);
    CheckParaUtils.checkDbObjectValidity(taskDTO, logger, String.format(DcsErrorCode.TASK_NULL.getDesc(), taskId));
    JobDTO jobDTO = jobService.selectByPrimaryKey(taskDTO.getJobId());
    CheckParaUtils.checkDbObjectValidity(jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), taskDTO.getJobId()));

    // 语法验证
//        boolean success = datasourceQueryService.checkGrammar(jobDTO.getSrcDatasourceId(), taskDTO.getSrcTableName(), filterCondition);
    boolean success = baseDbService.checkGrammar(jobDTO.getSrcDatasourceId(), taskDTO.getSrcTableName(), filterCondition);
    logger.info(String.format("语法验证结果：%s,数据源id：%s,表名：%s,过滤条件：%s", success, jobDTO.getSrcDatasourceId(),
            taskDTO.getSrcTableName(), filterCondition));
    return success;
  }

  /**
   * 创建表
   */
  @Transactional(rollbackFor = Exception.class)
  public synchronized Boolean createTable(String taskId) throws Exception {
    logger.info("************开始创建表,taskId={}", taskId);

    TaskDTO taskDTO = this.selectByPrimaryKey(taskId);
    CheckParaUtils.checkDbObjectValidity(taskDTO, logger, String.format(DcsErrorCode.TASK_NULL.getDesc(), taskId));
    JobDTO jobDTO = jobService.selectByPrimaryKey(taskDTO.getJobId());
    CheckParaUtils.checkDbObjectValidity(jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), taskDTO.getJobId()));

    TaskInterface taskInterface = taskFactory.getTaskInterface(jobDTO.getJobType());
    CheckParaUtils.checkObjectValidity(taskInterface, logger, DcsErrorCode.UNSUPPORTED_JOB_TYPE.getDesc() + jobDTO.getJobType());

    saveTableLog(taskDTO, "开始创建表");
    return taskInterface.createTable(jobDTO, taskDTO);
  }

  /**
   * 根据任务状态更新半结构化采集成功的文件
   */
  public synchronized Boolean updateTaskStatus(String taskInstanceId, Integer status) throws Exception {
    logger.info("************开始更新任务状态,taskInstanceId={},status={}", taskInstanceId, status);

    TaskInstanceDto taskInstanceDto = jobService.getTaskInstanceById(taskInstanceId);
    CheckParaUtils.checkDbObjectValidity(taskInstanceDto, logger, String.format("没有查到数据，请确保taskInstanceId:%s的任务实例是否存在",
            taskInstanceId));
    TaskDTO taskDTO = this.selectByPrimaryKey(taskInstanceDto.getName());
    CheckParaUtils.checkDbObjectValidity(taskDTO, logger, String.format(DcsErrorCode.TASK_NULL.getDesc(), taskInstanceDto.getName()));
    JobDTO jobDTO = jobService.selectByPrimaryKey(taskDTO.getJobId());
    CheckParaUtils.checkDbObjectValidity(jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), taskDTO.getJobId()));

    if (jobDTO.getJobType() == JobType.SEMISTRUCTURE.getCode() && status == LogStatus.COMPLETED.getCode()) {
      // 半结构化采集保存dcs_job_file数据
      String filePath = "";
      String defaultFS = "";

      JSONObject taskInstance = JSONObject.parseObject(taskInstanceDto.getTaskJson());
      JSONObject params = taskInstance.getJSONObject(DcsConstants.PARAMS);
      JSONObject json = params.getJSONObject(DcsConstants.JSON);
      JSONObject job = json.getJSONObject(DcsConstants.JOB);
      JSONArray contentArray = job.getJSONArray(DcsConstants.CONTENT);
      JSONObject content = contentArray.getJSONObject(0);
      JSONObject reader = content.getJSONObject(DcsConstants.READER);
      JSONObject parameter = reader.getJSONObject(DcsConstants.PARAMETER);
      defaultFS = parameter.getString(DcsConstants.DEFAULT_FS);

      if(StringUtils.isEmpty(defaultFS)){
        defaultFS = parameter.getString(DcsConstants.ENDPOINT) + Constants.SINGLE_SLASH;
      }

      JSONObject localParametersMap = params.getJSONObject(DcsConstants.LOCAL_PARAMETERS_MAP);
      JSONObject hdfsFile = localParametersMap.getJSONObject(ProcessConstants.HDFS_FILE_PATH + Constants.SUBTRACT_STRING + taskInstanceDto.getName());
      filePath = hdfsFile.getString(DcsConstants.VALUE);
      String realFilePath = defaultFS + filePath;

      JobFileDTO jobFileQuery = new JobFileDTO();
      jobFileQuery.setTaskId(taskInstanceDto.getName());
      jobFileQuery.setFilePath(realFilePath);
      JobFileDTO jobFileDTO = jobFileService.selectOne(jobFileQuery);
      if (jobFileDTO == null) {
        jobFileDTO = new JobFileDTO();
        jobFileDTO.setId(ComUtil.getId());
        jobFileDTO.setJobId(jobDTO.getId());
        jobFileDTO.setTaskId(taskInstanceDto.getName());
        jobFileDTO.setProcessInstanceId(taskInstanceDto.getProcessInstanceId());
        jobFileDTO.setFilePath(realFilePath);
        jobFileService.insert(jobFileDTO);
      } else {
        jobFileDTO.setProcessInstanceId(taskInstanceDto.getProcessInstanceId());
        jobFileService.updateByPrimaryKeySelective(jobFileDTO);
      }
      logger.info("jobId:{}, hdfsFilePath:{}", jobDTO.getId(), jobFileDTO.getFilePath());

      // 增量采集,记录文件时间
      updateFileDate(taskDTO, jobDTO, defaultFS, filePath);
    }

    // 推送给平台监控报警消息
    sendAlertMsg(jobDTO, status);

    return true;
  }

  /**
   * 推送给平台监控报警消息
   */
  private void sendAlertMsg(JobDTO jobDTO, Integer status){
    if(status == LogStatus.EXECUTE_FAIL.getCode()){
      AlertDTO alertDTOQuery = new AlertDTO();
      alertDTOQuery.setJobId(jobDTO.getId());
      alertDTOQuery.setAlertStatus(AlertStatus.ENABLE.getCode());
      List<AlertDTO> alertList = alertService.selectList(alertDTOQuery);
      if (CollectionUtils.isNotEmpty(alertList)) {
        for(AlertDTO alert : alertList){
          String userIds = alert.getUserIds();
          if(StringUtils.isNotEmpty(userIds)){
            String[] userArr = userIds.split(Constants.COMMA);

            Date date = new Date();
            // 调用平台接口发送消息
            for(String userId : userArr){
              try{
                if (StringUtils.isEmpty(userId)) {
                  continue;
                }
                SysMsgDTO msg = new SysMsgDTO();
                msg.setTitle(jobDTO.getName() + " 作业" + LogStatus.EXECUTE_FAIL.getDescp());
                msg.setContent(jobDTO.getName() + " 作业" + LogStatus.EXECUTE_FAIL.getDescp());
                msg.setSendUser(jobDTO.getCreatedBy());
                msg.setRecvUser(userId);
                msg.setSourceCode("personal");
                msg.setSourceName("数据采集通知");
                msg.setSendDate(date);
                msg.setMsgType("0");
                msg.setSendType("0");
                msg.setSysApplicationId(ThreadContextHelper.getAppId());
                sysMsgClient.sendmsg(msg);
              }catch (Exception e){
                logger.error("调用平台接口发送报警消息失败");
              }
            }
          }
        }
      }
    }
  }

  /**
   * 增量采集,记录文件时间
   * @param taskDTO
   */
  private void updateFileDate(TaskDTO taskDTO, JobDTO jobDTO, String defaultFS, String filePaths) throws Exception{
    if(taskDTO.getSyncMode() == SyncMode.DATAX_INCREMENT_SYNC.getCode()){
      String srcDatasourceId = jobDTO.getSrcDatasourceId();
      String systemBuilt = jobDTO.getExtraParams();
      String projectId = jobDTO.getProjectId();
      IFileAdapter fileAdapter = null;

      // 系统内置数据源
      if(Enable.ONE.getDescp().equals(systemBuilt)){
        fileAdapter = FileAdapterUtils.getFileAdapterByProject(projectId);
      }else{
        // 数据源中的数据源
        StorageResourceConf conf = dataSourceUtils.getStorageResourceByDatasourceId(srcDatasourceId);
        fileAdapter = FileAdapterUtils.getFileService(conf);
      }

      long recentlyTime = 0L;
      String[] arr = filePaths.split(Constants.COMMA);
      for(String filePath : arr){
        BdpFileStatus bdpFileStatus = fileAdapter.getFileStatus(defaultFS + filePath);
        if(bdpFileStatus != null){
          if(bdpFileStatus.getModificationTime() > recentlyTime){
            recentlyTime = bdpFileStatus.getModificationTime();
          }
        }
      }

      // 存syncTimePoint  Date类型字段，精度会丢失，故存放long类型字段
      taskDTO.setSyncIdValue(recentlyTime);
      this.updateByPrimaryKeySelective(taskDTO);
    }
  }

  /**
   * 获取待同步的文件
   *
   * @param taskId
   * @return
   * @throws Exception
   */
  public synchronized String getSyncFile(String taskId) throws Exception {
    logger.info("************开始获取待同步的文件,taskId={}", taskId);
    String hdfsFilePath = "";
    TaskDTO taskDTO = this.selectByPrimaryKey(taskId);
    CheckParaUtils.checkDbObjectValidity(taskDTO, logger, String.format(DcsErrorCode.TASK_NULL.getDesc(), taskId));
    JobDTO jobDTO = jobService.selectByPrimaryKey(taskDTO.getJobId());
    CheckParaUtils.checkDbObjectValidity(jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), taskDTO.getJobId()));

    hdfsFilePath = fileManageService.getSyncFile(taskDTO, jobDTO);

    logger.info("************获取待同步的文件,hdfsFilePath={}", hdfsFilePath);
    return hdfsFilePath;
  }
//    public synchronized String getSyncFile(String taskId) throws Exception {
//        logger.info("************开始获取待同步的文件,taskId={}", taskId);
//        String hdfsFilePath = "";
//        TaskDTO taskDTO = this.selectByPrimaryKey(taskId);
//        CheckParaUtils.checkDbObjectValidity(taskDTO, logger, StringUtils.format(DcsErrorCode.TASK_NULL.getDesc(), taskId));
//        JobDTO jobDTO = jobService.selectByPrimaryKey(taskDTO.getJobId());
//        CheckParaUtils.checkDbObjectValidity(jobDTO, logger, StringUtils.format(DcsErrorCode.JOB_NULL.getDesc(), taskDTO.getJobId()));
//
//        String catalogPath = jobDTO.getSelectedFile();
//        CatalogAndFileDto catalogAndFileDtoQuery = fileService.getCatalogAndFileTree(catalogPath, jobDTO.getSrcDatasourceId());
//        List<CatalogAndFileDto> fileList = catalogAndFileDtoQuery.getChildren();
//        if (CollectionUtils.isEmpty(fileList)) {
//            logger.info("目录：{}中没有待同步的文件", catalogPath);
//            return hdfsFilePath;
//        }
//
//        hdfsFilePath = this.getHdfsPath(hdfsFilePath, taskDTO, fileList, catalogPath);
//
//
//        logger.info("************获取待同步的文件,hdfsFilePath={}", hdfsFilePath);
//        return hdfsFilePath;
//    }

  /**
   * 获取要同步的hdfs文件路径
   *
   * @param hdfsFilePath    hdfsFilePath
   * @param taskDTO         taskDTO
   * @param catalogFileList catalogFileList
   * @return 同步的hdfs文件路径
   */
  private String getHdfsPath(String hdfsFilePath, TaskDTO taskDTO, List<CatalogAndFileDto> catalogFileList, String catalogPath) {
    logger.info("catalogPath：{}", catalogPath);

    // 1、查询目录模板关联表获取文件列表
    CatalogTemplateDTO catalogTemplateQuery = new CatalogTemplateDTO();
    catalogTemplateQuery.setCatalogPath(catalogPath);
    catalogTemplateQuery.setTemplateId(taskDTO.getSrcTableName());
    List<CatalogTemplateDTO> catalogTemplateList = catalogTemplateService.selectList(catalogTemplateQuery);
    List<String> filePathList = new ArrayList<>();
    if (CollectionUtils.isNotEmpty(catalogTemplateList)) {
      for (CatalogTemplateDTO catalogTemplate : catalogTemplateList) {
        filePathList.add(catalogTemplate.getFilePath());
      }
    }

    List<CatalogAndFileDto> fileList = new ArrayList<>();
    for (CatalogAndFileDto catalogAndFileDto : catalogFileList) {
      if (filePathList.contains(catalogAndFileDto.getPath())) {
        fileList.add(catalogAndFileDto);
      }
    }

    // 2、根据已经同步成功的文件，并结合任务关联的模板下所有文件，获取一个未同步的文件
    JobFileDTO jobFileDTOQuery = new JobFileDTO();
    jobFileDTOQuery.setTaskId(taskDTO.getId());
    List<JobFileDTO> jobFileDTOList = jobFileService.selectList(jobFileDTOQuery);
    if (CollectionUtils.isNotEmpty(jobFileDTOList)) {
      for (CatalogAndFileDto catalogAndFileDto : fileList) {
        boolean flag = true;
        for (JobFileDTO jobFileDTO : jobFileDTOList) {
          if (jobFileDTO.getFilePath().contains(catalogAndFileDto.getParentCatalogPath() + Constants.SINGLE_SLASH + catalogAndFileDto.getName())) {
//                    if ((catalogAndFileDto.getParentCatalogPath() + Constants.SINGLE_SLASH + catalogAndFileDto.getName())
//                            .equals(jobFileDTO.getFilePath())) {
            flag = false;
            break;
          }
        }

        if (flag) {
          hdfsFilePath = catalogAndFileDto.getPath();
          break;
        }
      }
    } else {
      if (fileList.size() > 0) {
        hdfsFilePath = fileList.get(0).getPath();
      }
    }
    return hdfsFilePath;
  }

  /**
   * 查询任务列表
   *
   * @param jobId
   * @return
   */
  public List<TaskDTO> getTaskListByJobId(String jobId) {
    TaskDTO taskQuery = new TaskDTO();
    taskQuery.setJobId(jobId);
    taskQuery.setFlag(Enable.ZERO.getCode());
    return this.selectList(taskQuery);
  }

  // 目标列按源列顺序排列
  public List<ColumnInfo> sortBySrcColumn(List<ColumnInfo> srcColumnList, List<ColumnInfo> dstColumnList, String mappingColumn) {
    List<ColumnMappingDto> columnMappingDtoList = JSONUtils.toList(mappingColumn, ColumnMappingDto.class);
    List<ColumnInfo> columnList = new ArrayList<>();
    List<ColumnInfo> newSrcColumnList = new ArrayList<>();

    for (ColumnInfo srcColumn : srcColumnList) {
      if (CollectionUtils.isNotEmpty(columnMappingDtoList)) {
        for (ColumnMappingDto columnMappingDto : columnMappingDtoList) {
          if (srcColumn.getName().equalsIgnoreCase(columnMappingDto.getSrcColumnName())) {
            newSrcColumnList.add(srcColumn);
            break;
          }
        }
      }
    }

    for (ColumnInfo srcColumn : srcColumnList) {
      if (!newSrcColumnList.contains(srcColumn)) {
        newSrcColumnList.add(srcColumn);
      }
    }

    srcColumnList.clear();
    srcColumnList.addAll(newSrcColumnList);

    // 有映射关系的排在前面，无映射关系的排后面
    for (ColumnInfo srcColumn : srcColumnList) {
      String srcColumnName = srcColumn.getName();
      String dstColumnName = "";

      if (CollectionUtils.isNotEmpty(columnMappingDtoList)) {
        for (ColumnMappingDto columnMappingDto : columnMappingDtoList) {
          if (srcColumnName.equalsIgnoreCase(columnMappingDto.getSrcColumnName())) {
            dstColumnName = columnMappingDto.getDstColumnName();
            break;
          }
        }
      }

      if (StringUtils.isNotEmpty(dstColumnName)) {
        if (CollectionUtils.isNotEmpty(dstColumnList)) {
          for (ColumnInfo dstColumn : dstColumnList) {
            if (dstColumnName.equalsIgnoreCase(dstColumn.getName())) {
              columnList.add(dstColumn);
              break;
            }
          }
        }
      }
    }

    for (ColumnInfo dstColumn : dstColumnList) {
      if (!columnList.contains(dstColumn)) {
        columnList.add(dstColumn);
      }
    }
    return columnList;
  }

  /**
   * 前置处理
   */
  @Transactional(rollbackFor = Exception.class)
  public Boolean preProcess(String taskId) throws Exception {
    logger.info("************开始前置处理,taskId={}", taskId);

    TaskDTO taskDTO = this.selectByPrimaryKey(taskId);
    CheckParaUtils.checkDbObjectValidity(taskDTO, logger, String.format(DcsErrorCode.TASK_NULL.getDesc(), taskId));
    JobDTO jobDTO = jobService.selectByPrimaryKey(taskDTO.getJobId());
    CheckParaUtils.checkDbObjectValidity(jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), taskDTO.getJobId()));

    // 更新bdp_dcs_task_data_detail表预估总数字段(立即和定时都会调用)
    TaskDataDetailDTO taskDataDetailDTOQuery = TaskDataDetailDTO.builder().taskId(taskDTO.getId()).build();
    TaskDataDetailDTO taskDataDetailDTO = taskDataDetailService.selectOne(taskDataDetailDTOQuery);
    if (taskDataDetailDTO != null) {

      if (jobDTO.getJobType() == JobType.SEMISTRUCTURE.getCode() ||
              jobDTO.getJobType() == JobType.STREAM.getCode() ||
              jobDTO.getJobType() == JobType.API_HTTP.getCode()) {
        taskDataDetailDTO.setEstimateTotal(null);
      }else if(jobDTO.getJobType() == JobType.TRANSFER.getCode()){
        taskDataDetailDTO.setEstimateTotal(baseDbService.countTableSize(jobDTO.getSrcDatasourceId(),
                taskDataDetailDTO.getSrcTableName(), taskDTO.getFilterCondition()));
      }else if(jobDTO.getJobType() == JobType.SYNC.getCode() ){
        if(taskDTO.getSyncMode() != null && taskDTO.getSyncMode() == SyncMode.TOTAL_SYNC.getCode()){
          taskDataDetailDTO.setEstimateTotal(baseDbService.countTableSize(jobDTO.getSrcDatasourceId(),
                  taskDataDetailDTO.getSrcTableName(), taskDTO.getFilterCondition()));
        }
      }else{
        logger.warn("不支持的任务类型:{}", jobDTO.getJobType());
      }

      taskDataDetailService.updateByPrimaryKeySelective(taskDataDetailDTO);
      logger.info("更新bdp_dcs_task_data_detail表预估总数成功");
    }

    return true;
  }

  /**
   * 数据预览
   */
  public Map<String, Object> preview(String jobId, String srcTableName, Integer currentPage, Integer pageSize) throws Exception {
    Map<String, Object> map = new HashMap<>(8);

    JobDTO job = jobService.selectByPrimaryKey(jobId);
    CheckParaUtils.checkDbObjectValidity(job, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobId));

    TaskDTO task = jobService.selectByJobIdAndSrcTableName(jobId, srcTableName);

    CheckParaUtils.checkDbObjectValidity(task, logger, String.format("没有查到数据，请确保jobId：%s,srcTableName:%s的任务是否存在",
            jobId, srcTableName));

    try {
      map = getDataListFromDB(job, task, currentPage, pageSize, null);
      List<Map<String, Object>> resultList = (List<Map<String, Object>>) map.get("resultList");
      CommonUtil.filterMapNull(resultList);
    } catch (Exception e) {
      logger.error("数据预览异常", e);
      throw e;
    }

    return map;
  }

  /**
   * 从数据库实时查询数据
   *
   * @return 查询数据
   */
  public Map<String, Object> getDataListFromDB(JobDTO job, TaskDTO task, Integer currentPage, Integer pageSize, Map<String, Object> parameterMap) {
    Map<String, Object> map = new HashMap<>(8);
    List<BdpColumnDTO> columnList = new ArrayList<>();
    BdpTableDTO tableDTO = new BdpTableDTO();
    tableDTO.setDataSourceId(job.getDstDatasourceId());

    List<ColumnMappingDto> columnInfoDtoList = new ArrayList<>();
    if (JobType.STREAM.getCode() == job.getJobType()) {
      List<StreamColumnMappingDto> streamColumnMappingList = JSONUtils.toList(task.getMappingColumn(), StreamColumnMappingDto.class);
      if (CollectionUtils.isNotEmpty(streamColumnMappingList)) {
        for (StreamColumnMappingDto streamColumnMapping : streamColumnMappingList) {
          if (streamColumnMapping.getTemplateName().equals(task.getSrcTableName())) {
            columnInfoDtoList = streamColumnMapping.getColumnList();
          }
        }
      }
    } else {
      columnInfoDtoList = JSONUtils.toList(task.getMappingColumn(), ColumnMappingDto.class);
    }

    // 无字段直接返回
    if (CollectionUtils.isEmpty(columnInfoDtoList)) {
      return map;
    }

    for (ColumnMappingDto columnInfo : columnInfoDtoList) {
      columnInfo.setDstColumnName(columnInfo.getDstColumnName().toLowerCase());
      BdpColumnDTO column = new BdpColumnDTO();
      column.setName(columnInfo.getDstColumnName());
      column.setOrderBy(1);
      columnList.add(column);
    }
    tableDTO.setColumnList(columnList);

    DataSourceDTO datasource = dataSourceService.selectByPrimaryKey(job.getDstDatasourceId());
    String sql = "select * from " + DBUtils.convertTableName(datasource.getDatasourceType(), task.getDstTableName());
    logger.info("sql = {}", sql);

    map = dataQueryService.getDataListByTable(tableDTO, null, sql, parameterMap, currentPage, pageSize);
    map.put("columnList", columnInfoDtoList);
    map.remove("modelInfo");
    return map;
  }

  public boolean isExistTable(String datasourceId, String tableName) throws Exception {
    boolean existDstTable = false;
    List<String> tableList = baseDbService.getTableList(datasourceId);
    if (CollectionUtils.isNotEmpty(tableList)) {
      if (tableList.contains(tableName) || tableList.contains(tableName.toLowerCase())
              || tableList.contains(tableName.toUpperCase())) {
        existDstTable = true;
      }
    }
    return existDstTable;
  }

  /**
   * 根据是否有历史任务，判断是否为第一次执行job
   *
   * @return
   * @throws Exception
   */
  public boolean isCoverTable(JobDTO jobDTO, TaskDTO taskDTO) throws Exception {
    boolean isCoverTableFlag = true;
    if (jobDTO.getJobType() == JobType.SYNC.getCode() && taskDTO.getSyncMode() == SyncMode.DATAX_INCREMENT_SYNC.getCode()) {
      int num = jobService.countProcessInstanceByProcessDefinitionId(jobDTO.getId());
      if (num > 1) {
        isCoverTableFlag = false;
      }
    }
    return isCoverTableFlag;
  }

  /**
   * 是否需创建表
   *
   * @return
   * @throws Exception
   */
  private boolean isCreateTable(JobDTO jobDTO, String dstTable) throws Exception {
    boolean existDstTable = isExistTable(jobDTO.getDstDatasourceId(), dstTable);

    boolean createTable = false;
    if (jobDTO.getJobType() == JobType.TRANSFER.getCode()) {
      if (jobDTO.getTransferType() != null && jobDTO.getTransferType().contains(Enable.ONE.getDescp())) {
        createTable = true;
      }
    } else if (jobDTO.getJobType() == JobType.SYNC.getCode()) {
      if (existDstTable == false ||
              (existDstTable && jobDTO.getProcessMode() != ProcessMode.RETAIN_DST_TABLE.getCode())) {
        createTable = true;
      }
    } else if (jobDTO.getJobType() == JobType.SEMISTRUCTURE.getCode()) {
      if (jobDTO.getProcessMode() != ProcessMode.RETAIN_DST_TABLE.getCode()) {
        createTable = true;
      }
    } else if (jobDTO.getJobType() == JobType.STREAM.getCode()) {
      if (jobDTO.getProcessMode() != ProcessMode.RETAIN_DST_TABLE.getCode()) {
        createTable = true;
      }
    } else if (jobDTO.getJobType() == JobType.API_HTTP.getCode()) {
      if (existDstTable == false ||
              (existDstTable && jobDTO.getProcessMode() != ProcessMode.RETAIN_DST_TABLE.getCode())) {
        createTable = true;
      }
    } else {
      logger.warn("不支持的job类型:{}", jobDTO.getJobType());
    }

    return createTable;
  }

  /**
   * 删除任务
   */
  public void deleteTask(String jobId, String srcTableName) throws Exception {
    JobDTO jobDTO = jobService.selectByPrimaryKey(jobId);
    CheckParaUtils.checkDbObjectValidity(jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobId));

    TaskDTO query = new TaskDTO();
    query.setJobId(jobId);
    query.setFlag(Enable.ZERO.getCode());
    query.setSrcTableName(srcTableName);
    TaskDTO task = selectOne(query);
    CheckParaUtils.checkDbObjectValidity(task, logger, String.format("没有查到数据，请确保jobId：%s,srcTableName:%s的任务是否存在", jobId, srcTableName));
    delete(task);
    BdpLogUtil.log4Delete(task);

    // 修改job表selectedTable
    TableMappingDto table = null;
    List<TableMappingDto> tableMappingList = JSONUtils.toList(jobDTO.getSelectedTable(), TableMappingDto.class);
    for (TableMappingDto tableMapping : tableMappingList) {
      if (tableMapping.getSrc_table().equals(srcTableName)) {
        table = tableMapping;
        break;
      }
    }

    tableMappingList.remove(table);

    jobDTO.setSelectedTable(JSONArray.toJSONString(tableMappingList));
    jobService.updateByPrimaryKeySelective(jobDTO);
  }

  /**
   * 根据jobId获取源表
   * @param jobId
   * @return
   * @throws Exception
   */
  public List<String> getSrcTableListByJobId(String jobId) throws Exception {
    TaskDTO taskQuery = new TaskDTO();
    taskQuery.setJobId(jobId);
    taskQuery.setFlag(Enable.ZERO.getCode());
    List<TaskDTO> taskDTOList = this.selectList(taskQuery);
    List<String> srcTableList = taskDTOList.stream().map(TaskDTO::getSrcTableName).collect(Collectors.toList());

    return srcTableList;
  }

  /**
   * 列名称转换
   *
   * @return
   */
  public Map<String, String> convertColumnName(TaskDTO taskDTO) {
    Map<String, String> columnMap = new HashMap<>(16);
    List<ColumnMappingDto> columnMappingDtoList = JSONUtils.toList(taskDTO.getMappingColumn(), ColumnMappingDto.class);
    if (CollectionUtils.isNotEmpty(columnMappingDtoList)) {
      for (ColumnMappingDto columnMappingDto : columnMappingDtoList) {
        columnMap.put(columnMappingDto.getSrcColumnName(), columnMappingDto.getDstColumnName());
      }
    }
    return columnMap;
  }

  /**
   * 回写创建表sql语句
   */
  public Boolean updateSql(Map<String, String> map, TaskDTO taskDTO) throws Exception {
    this.saveTableLog(taskDTO, String.format("创建表sql:%s", map.get(Constants.CREATE_TABLE_SQL)));

    boolean createTableFlag = true;

    TaskTableDetailDTO query = TaskTableDetailDTO.builder().taskId(taskDTO.getId()).srcTableName(taskDTO.getSrcTableName()).build();
    TaskTableDetailDTO tableDetailDTO = taskTableDetailService.selectOne(query);
    CheckParaUtils.checkDbObjectValidity(tableDetailDTO, logger, String.format("任务id:%s,没找到对应的表结构迁移明细数据", taskDTO.getId()));

    tableDetailDTO.setCreateSql(map.get(Constants.CREATE_TABLE_SQL));
    if (Boolean.valueOf(map.get(Constants.SUCCESS))) {
      tableDetailDTO.setTaskStatus(TaskDetailStatus.COMPLETED.getCode());

      taskDTO.setIsCreated(Enable.ONE.getCode());
      this.updateByPrimaryKeySelective(taskDTO);
      logger.info("表:{} 更新为已创建", taskDTO.getDstTableName());

      this.saveTableLog(taskDTO, String.format("创建表成功"));
    } else {
      tableDetailDTO.setTaskStatus(TaskDetailStatus.EXECUTE_FAIL.getCode());
      createTableFlag = false;

      this.saveTableLog(taskDTO, String.format("创建表失败,原因是:%s", map.get(Constants.MESSAGE)));
    }
    taskTableDetailService.updateByPrimaryKeySelective(tableDetailDTO);
    logger.info("创建表sql={}", map.get(Constants.CREATE_TABLE_SQL));
    logger.info("************创建表返回结果：{},状态为:{},表名:{}", map.get(Constants.SUCCESS), tableDetailDTO.getTaskStatus(), taskDTO.getDstTableName());

    return createTableFlag;
  }

  public void saveTableLog(TaskDTO task, String tableLog) {
    tableLogService.saveTableLog(task, tableLog);
  }

  /**
   * 删除增量时间
   * @param jobId
   */
  public void clearSyncTimePointByJobId(String jobId) {
    TaskDTO taskQuery = new TaskDTO();
    taskQuery.setJobId(jobId);
    taskQuery.setFlag(Enable.ZERO.getCode());
    List<TaskDTO> list = this.selectList(taskQuery);

    if (CollectionUtils.isNotEmpty(list)) {
      for(TaskDTO task : list){
        task.setSyncTimePoint(null);
        this.updateByPrimaryKeySelective(task);
      }
    }
  }

}